package com.shao.heartbeatDetector;

import com.shao.Configuration;
import com.shao.SrpcBootstrap;
import com.shao.channelHandler.enums.RequestType;
import com.shao.compress.CompressorFactory;
import com.shao.discovery.NettyBootstrapInitializer;
import com.shao.discovery.Registry;
import com.shao.serialize.SerializerFactory;
import com.shao.transport.message.SrpcRequest;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author PaoFu
 * @version 1.0
 * @data 2025/5/8 11:43
 */
@Slf4j
public class HeartbeatDetection {

    // 心跳检测失败尝试重新连接的次数
    private static final int TRY_AGAIN_FREQUENCY = 3;

    private static final Configuration configuration = SrpcBootstrap.getInstance().getConfiguration();

    public static void heartbeatDetection(String serviceName) {
        // 拉取服务列表
        Registry registry = configuration.getRegistryConfig().getRegistry();
        List<InetSocketAddress> addresses = registry.lookup(serviceName);
        for (InetSocketAddress address : addresses) {
            // 建立链接
            if (SrpcBootstrap.SOCKET_CHANNEL.get(address) == null) {
                try {
                    Channel channel = NettyBootstrapInitializer.getBootstrap().connect(address).sync().channel();
                    SrpcBootstrap.SOCKET_CHANNEL.put(address, channel);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            Thread thread = new Thread(() -> new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 2000), "srpc-HeartbeatDetection-Thread");
            // 设置守护线程
            thread.setDaemon(true);
            thread.start();
        }
    }

    private static class MyTimerTask extends TimerTask {
        @Override
        public void run() {
            SrpcBootstrap.SHORTEST_RESPONSE_TIME_CHANNEL.clear();
            Map<InetSocketAddress, Channel> addressChannelMap = SrpcBootstrap.SOCKET_CHANNEL;
            addressChannelMap.keySet().forEach(key -> {
                // 创建请求
                SrpcRequest srpcRequest = SrpcRequest.builder()
                        .requestId(configuration.getIdGenerator().getId())
                        .compressType(CompressorFactory.getCompressor(configuration.getCompressType()).getCode())
                        .requestType(RequestType.HEART_REQUEST.getType())
                        .serializeType(SerializerFactory.getSerializer(configuration.getSerializeType()).getCode())
                        .build();
                boolean flag = performPipeConn(key, addressChannelMap, srpcRequest);
                if (!flag) {
                    // 超时重试
                    for (int i = 0; i < TRY_AGAIN_FREQUENCY; i++) {
                        try {
                            Thread.sleep(new Random().nextInt((i + 1) * 10));
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        boolean tryAgainFlag = performPipeConn(key, addressChannelMap, srpcRequest);
                        if (tryAgainFlag) {
                            // 成功则直接跳出循环
                            break;
                        }
                        log.error("服务【{}】心跳检测正在尝试第【{}】次重新连接", key, (i + 1));
                        if ((i == TRY_AGAIN_FREQUENCY - 1)) {
                            // 尝试3次还没连接成功则判定该服务已经下线
                            log.warn("服务【{}】已下线，正在从服务列表剔除该服务", key);
                            SrpcBootstrap.SOCKET_CHANNEL.remove(key);
                        }
                    }
                }
            });
        }
    }

    private static boolean performPipeConn(InetSocketAddress key, Map<InetSocketAddress, Channel> addressChannelMap, SrpcRequest srpcRequest) {
        Channel channel = addressChannelMap.get(key);
        CompletableFuture<Object> completableFuture = new CompletableFuture<>();
        // 暴露 CompletableFutrue
        SrpcBootstrap.PENDING_REQUEST.put(srpcRequest.getRequestId(), completableFuture);
        long startTime = System.currentTimeMillis();
        try {
            if (channel != null) {
                channel.writeAndFlush(srpcRequest).get(1, TimeUnit.SECONDS);
            }
        } catch (InterruptedException | ExecutionException e) {
            log.error("心跳检测时发现服务异常关闭");
            return false;
        } catch (TimeoutException e) {
            log.error("服务【{}】的心跳检测获取连接超时", key);
            return false;
        }
        long responseTime = System.currentTimeMillis() - startTime;
        log.debug("和【{}】服务器的响应时间为：【{}】", key, responseTime);
        SrpcBootstrap.SHORTEST_RESPONSE_TIME_CHANNEL.put(responseTime, channel);
        return true;
    }
}
